//var Rx = require('rx');
//var RxNode = require('rx-node');
//
//var source = Rx.Observable.return(42);
//
//var emitter = RxNode.toEventEmitter(source, 'data');
//
//emitter.on('data', function (data) {
//  console.log('Data: ' + data);
//});
//
//emitter.on('end', function () {
//  console.log('End');
//});
//
//// Ensure to call publish to fire events from the observable
//emitter.publish();


var RxNode = require('rx-node');

//var subscription = RxNode.fromReadableStream(process.stdin)
//  .subscribe(function (x) { console.log(x); });

var Readable = require('stream').Readable;

var rs = new Readable;
rs.push('beep ');
rs.push('boop\n');
rs.push(null);
//rs.pipe(process.stdout);

var subscription = RxNode.fromStream(rs, 'end')
  .subscribe(function (x) { console.log(x); });


var EventEmitter = require('events').EventEmitter;
var event = new EventEmitter();
event.on('some_event', function() {
  console.log('some_event 事件触发');
});

var fetch = require('isomorphic-fetch');

var mongodb = require('mongodb');

var server = new mongodb.Server('localhost', 27017, {
  auto_reconnect: true
});

//连接的数据库为stock
var db = new mongodb.Db('stocks', server, {
  safe: true
});

db.open(function(err, db) {
  var basicsCollection = db.collection('stock_basics');

  basicsCollection.find({pe: {$lte: 15}}, {limit: 1000}).toArray(function (err, doc) {
    setInterval (function () {
      var list = doc.map(function (item) {
        var code = item._id
        if (/^0|3/.test(code)) {
          code = 'sz' + code
        }
        if (/^6/.test(code)) {
          code = 'sh' + code
        }
        return code
      })

      list.forEach(item => {
        var dayInfo = db.collection(item._id)
        dayInfo.find({}, {limit: 100}).toArray(function (err, doc) {

        })
      })

      fetch('http://hq.sinajs.cn/list=' + list.join(','))
        .then(function (res) {
          return res.text()
        })
        .then(function(data) {
          console.log(data)
          event.emit(data);
        });
    }, 1000)
  })
})
